package com.example.config;

import com.example.util.MQConsumeMsgListenerProcessor;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.PropertySource;

@Configuration
public class MQConsumeConfig {

    public static final Logger LOGGER = LoggerFactory.getLogger(MQConsumeConfig.class);
    @Value("${rocketmq.consumer.namesrvAddr}")
    private String namesrvAddr;
    @Value("${rocketmq.consumer.groupName}")
    private String groupName;
    @Value("${rocketmq.consumer.topics}")
    private String topics;
    @Autowired
    MQConsumeMsgListenerProcessor mqConsumeMsgListenerProcessor;

    @Bean
    public DefaultMQPushConsumer createConsumer() {
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer(this.groupName);
        defaultMQPushConsumer.setNamesrvAddr(this.namesrvAddr);
        defaultMQPushConsumer.registerMessageListener(mqConsumeMsgListenerProcessor);
        try {
            String[] topicTagsArr = topics.split(";");
            for (String topicTags : topicTagsArr) {
                String[] topicTag = topicTags.split("~");
                defaultMQPushConsumer.subscribe(topicTag[0],topicTag[1]);
            }
            defaultMQPushConsumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        return defaultMQPushConsumer;
    }
}
